package com.netty.admin.consumer;

import com.alibaba.fastjson.JSON;
import com.netty.admin.service.MessageService;
import com.netty.common.constant.MqConstants;
import com.netty.common.entity.mq.MqMessage;
import com.netty.common.enums.ExceptionEnum;
import com.netty.common.util.RedisUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.Objects;

/**
 * @author LiuXiaoYu
 * @date 2022/2/6- 16:02
 */
@Service
@Slf4j
public class MessageConsumer {
    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private MessageService messageService;

    /**
     * 消费消息
     */
    @RabbitListener(queues = MqConstants.NETTY_MESSAGE_QUEUE)
    @Transactional(rollbackFor = Exception.class)
    public void consumerOrderMsg(Message message, Channel channel) throws IOException {
        log.info("消费者开始消费消息message：{}", message);
        //通过message唯一ID来防止重复消费：为什么需要防止重复消费，消费端没有在有效时间内确认消息已消费
        //1 - 获取message唯一ID
        String messageId = message.getMessageProperties().getMessageId();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //redis可能因为网络原因导致或者真的被删除了 报空指针 应该用数据库的
        MqMessage mqMessage=(MqMessage)redisUtil.get(MqConstants.NETTY_MESSAGE_LOG_KEY+messageId);
        if (Objects.nonNull(mqMessage)){
            //2 - 通过redis查询是否存在该消息ID，存在则不再消费
            if (Objects.nonNull(messageId)&& !redisUtil.hasKey(MqConstants.NETTY_CONSUMED_KEY+messageId)) {
                String msg = new String(message.getBody());
                try {
                    com.netty.common.entity.Message msgToDataBase = JSON.parseObject(msg, com.netty.common.entity.Message.class);
                    messageService.getBaseMapper().insert(msgToDataBase);
                    channel.basicAck(deliveryTag, false);

                    mqMessage.setMessageId(messageId).setStatus(MqConstants.MESSAGE_CONSUMERED);
                    //更改msg状态 - 消费成功
                    boolean consumerResult=redisUtil.set(MqConstants.NETTY_MESSAGE_LOG_KEY + messageId, mqMessage);
                    log.info("redis里面的mq消息日志执行完成了");
                    if (consumerResult) {
                        //3 - 消费完成，在redis 中设置该key，value为true
                        redisUtil.set(MqConstants.NETTY_CONSUMED_KEY+messageId, true);

                    }
                    log.info("消费结果：消息签收成功，消息唯一ID：{},更改数据库消息为已消费状态", messageId);
                } catch (Exception e) {
                    log.error("消息消费异常：{}", e.getMessage(),e);
                    //拒绝签收此消息不用重新入队 如果配置死信队列 则会丢进死信队列
                    channel.basicNack(deliveryTag, false,false);
                    mqMessage.setMessageId(messageId).setStatus(MqConstants.MESSAGE_CONSUMERED_FAIL);
                    redisUtil.set(MqConstants.NETTY_MESSAGE_LOG_KEY + messageId,mqMessage);
                    log.error("消费结果：消息签收失败，消息唯一ID：{},更改数据库消息为消费失败状态", messageId);
                }
            } else {
                log.error("该消息已经被消费：{},请勿重新消费,丢进死信队列", messageId);
                //重复消费的消息也会直接丢进死信队列
                channel.basicNack(deliveryTag,false,false);
            }
        }else {
            log.error("存储在redis中的消息日志丢失,id:"+messageId);
            log.error("直接放进死信队列消费,id:"+messageId);
            //直接拒收消息
            channel.basicNack(deliveryTag, false,false);
        }
    }

    /**二次重复消费的消息以及发生异常catch掉的消息 都会丢进死信队列
     * 监听死信队列
     * @param message 消息
     * @param channel 通道
     */
    @RabbitListener(queues = MqConstants.NETTY_MESSAGE_DEAD_QUEUE)
    @Transactional(rollbackFor = Exception.class)
    public void consumerMailDeadMessage(Message message,Channel channel) throws IOException {
        //1 - 获取message唯一ID
        String messageId = message.getMessageProperties().getMessageId();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            channel.basicAck(deliveryTag,false);
            log.info("监听到死信队列消息,消息Id:{},消费成功",messageId);
        }catch (Exception e){
            log.error("死信队列消息异常,消息id:{},消费失败",messageId);
            channel.basicNack(deliveryTag,false,false);
        }
    }
}
